package org.yuanzheng.window

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.yuanzheng.source.StationLog

/**
 * @author yuanzheng
 * @date 2020/6/21-17:18
 */
object TestAggregateFunction {
  def main(args: Array[String]): Unit = {
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    import org.apache.flink.streaming.api.scala._

    //读取数据源
    val stream: DataStream[StationLog] = streamEnv.socketTextStream("192.168.1.8", 8888)
      .map(line => {
        var split = line.split(",")
        new StationLog(split(0).trim, split(1).trim, split(2).trim, split(3).trim, split(4).trim.toLong, split(5).trim.toLong)
      })

    //开窗
    stream.map(log => (log.sid, 1)).keyBy(_._1)
      .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
      .aggregate(new myAggregateFunction, new myWindowFunction).print()
    streamEnv.execute()
  }

  //add方法每条数据都会执行，getResult方法在窗口结束时执行
  class myAggregateFunction extends AggregateFunction[(String, Int), Long, Long] {
    override def createAccumulator(): Long = 0
    override def add(in: (String, Int), acc: Long): Long = acc + in._2
    override def getResult(acc: Long): Long = acc
    override def merge(acc: Long, acc1: Long): Long = acc + acc1
  }

  //windowFunction输入数据来源于AggregateFunction，窗口结束时先执行aggregateFunction的getResult后执行apply方法
  class myWindowFunction extends WindowFunction[Long, (String, Long), String, TimeWindow] {
    override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[(String, Long)]): Unit = {
      out.collect((key, input.iterator.next()))
    }
  }
}
